package cn.kgc.gmall.app.func;

import cn.kgc.gmall.common.GmallConfig;
import cn.kgc.gmall.utils.DimUtil;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.*;
import java.util.Collection;
import java.util.Set;

/**
 * 自定义sink
 */
public class DimSink extends RichSinkFunction<JSONObject> {

    Connection conn = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        //初始化Phoenix连接
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        conn = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
    }

    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
        super.invoke(value, context);
        //{"database":"gmall_2022","table":"","type":"delete","ts":1661916124,"xid":99635,"xoffset":46458,"data":{"id":1564815757992218659,"user_id":1200}}
        // 获取表名
        String table = value.getString("sink_table");

        JSONObject data = value.getJSONObject("data");
        if (data != null && data.size() > 0){
            // 获取sql语句
            String sql = genUpsertSql(table.toUpperCase(),data);
            PreparedStatement ps = null;
            try {
                ps = conn.prepareStatement(sql);
                System.out.println("往hbase添加数据:"+sql);
                ps.executeUpdate();
                // System.out.println("执行了更新");
                // 当执行对hbase中的数据进行修改的时候同时清除redis中的缓存 更新和删除操作才需要缓存清除
                if (
                        value.getString("type").equals("update") ||
                        value.getString("type").equals("delete")
                ){

                    DimUtil.deleteCached(table,data.getString("id"));
                }
                conn.commit();
            } catch (SQLException e) {
                throw new RuntimeException(e);
            }finally {
                if( ps != null){
                    ps.close();

                }
            }
        }


    }

    /**
     * 拼接upsert语句 返回成一个字符串
     * @param table
     * @param data
     * @return
     */
    private String genUpsertSql(String table, JSONObject data) {
        Set<String> keys = data.keySet();
        Collection<Object> values = data.values();
        // upsert into student(字段名,字段名,...) values('1001','zhangsan','beijing',...);
        StringBuffer sql = new StringBuffer(" upsert into "+GmallConfig.HBASE_SCHEMA+"."+table+" ( ");
        sql.append(StringUtils.join(keys,","));
        sql.append(" ) values ('");
        sql.append(StringUtils.join(values,"','"));
        sql.append("' ) ");
        //System.out.println("sql:"+sql);
        return sql.toString();
    }
}
